/*
 * Copyright (c) Meta Platforms, Inc. and affiliates.
 *
 * This software may be used and distributed according to the terms of the
 * GNU General Public License version 2.
 */

use std::clone::Clone;
use std::collections::HashMap;
use std::collections::HashSet;
use std::sync::Arc;

use anyhow::Context;
use anyhow::Result;
use anyhow::anyhow;
use anyhow::ensure;
use async_recursion::async_recursion;
use blobstore::Storable;
use cloned::cloned;
use context::CoreContext;
use derivative::Derivative;
use either::Either;
use either::Either::*;
use futures::stream;
use futures::stream::StreamExt;
use futures::stream::TryStreamExt;
use futures_stats::TimedFutureExt;
use manifest::BonsaiDiffFileChange;
use maplit::hashmap;
use mononoke_types::BlobstoreValue;
use mononoke_types::BonsaiChangesetMut;
use mononoke_types::ChangesetId;
use mononoke_types::ContentId;
use mononoke_types::FileChange;
use mononoke_types::FileContents;
use mononoke_types::FileType;
use mononoke_types::GitLfs;
use mononoke_types::NonRootMPath;
use mononoke_types::RepositoryId;
use mononoke_types::TrackedFileChange;
use mononoke_types::hash::GitSha1;
use reporting::log_debug;
use reporting::log_info;
use scuba_ext::FutureStatsScubaExt;
use sorted_vector_map::SortedVectorMap;

use crate::git_submodules::in_memory_repo::InMemoryRepo;
use crate::git_submodules::utils::build_recursive_submodule_deps;
use crate::git_submodules::utils::get_git_hash_from_submodule_file;
use crate::git_submodules::utils::get_submodule_bonsai_changeset_id;
use crate::git_submodules::utils::get_submodule_file_content_id;
use crate::git_submodules::utils::get_submodule_repo;
use crate::git_submodules::utils::get_x_repo_submodule_metadata_file_path;
use crate::git_submodules::utils::is_path_git_submodule;
use crate::git_submodules::utils::list_non_submodule_files_under;
use crate::git_submodules::utils::submodule_diff;
use crate::types::Repo;
use crate::types::SubmoduleExpansionContentIds;
use crate::types::SubmodulePath;

// TODO(T174902563): support expansion of git submodules
/// Everything needed to expand submodule changes
#[derive(Clone, Derivative)]
#[derivative(Debug)]
pub struct SubmoduleExpansionData<'a, R: Repo> {
    // Submodule dependencies of from the small repo, which have to be loaded
    // and available to (a) expand submodule file changes or (b) validate
    // that a bonsai in the large repo doesn't break the consistency of submodule
    // expansions.
    #[derivative(Debug = "ignore")]
    pub submodule_deps: &'a HashMap<NonRootMPath, Arc<R>>,
    pub x_repo_submodule_metadata_file_prefix: &'a str,
    /// Used to determine the direction of the sync and log relevant information
    /// to scuba.
    pub small_repo_id: RepositoryId,
    /// Read-only version of the large repo, which performs any writes in memory.
    /// This is needed to validate submodule expansion in large repo bonsais.
    #[derivative(Debug = "ignore")]
    pub large_repo: InMemoryRepo,
    /// List git commit hashes that are known dangling submodule pointers in the
    /// repo's history, i.e. don't actually exist in the submodule repo it's
    /// supposed to point to.
    /// This can happen after non-fast-forward pushes or accidentally pushing
    /// commits with local submodule pointers.
    ///
    /// The expansion of these commits will contain a single text file informing
    /// that the expansion belongs to a dangling submodule pointer.
    pub dangling_submodule_pointers: Vec<GitSha1>,
}

/// Used to distinguish file changes that came from the original bonsai or
/// were generated by the submodule expansion logic.
/// This is needed to make sure that if the original bonsai makes changes to
/// a path that the expansion logic generates changes, the change from the
/// original bonsai always takes precedence.
///
/// Example: user deletes submodule and keeps its static copy as a regular
/// directory in the same commit. The expansion logic generates a deletion
/// for the entire expansion, but the user added the files as regular changes,
/// so they should be part of the bonsai.
enum ExpansionFileChange {
    /// File change from the original small repo bonsai
    Original((NonRootMPath, FileChange)),
    /// File change generated by submodule expansion primitives
    Generated((NonRootMPath, FileChange, SubmodulePath)),
}

/// Iterate over all file changes from the bonsai being synced and expand any
/// changes to git submodule files, generating the bonsai that will be synced
/// to the large repo.
pub(crate) async fn expand_all_git_submodule_file_changes<'a, R: Repo>(
    ctx: &'a CoreContext,
    cs: BonsaiChangesetMut,
    small_repo: &'a R,
    sm_exp_data: SubmoduleExpansionData<'a, R>,
) -> Result<(BonsaiChangesetMut, SubmoduleExpansionContentIds)> {
    let fcs: SortedVectorMap<NonRootMPath, FileChange> = cs.file_changes;
    let parents = cs.parents.as_slice();

    let (orig_fcs, mut expanded_fcs, submodule_expansion_content_ids): (
        SortedVectorMap<NonRootMPath, FileChange>,
        SortedVectorMap<NonRootMPath, FileChange>,
        SubmoduleExpansionContentIds,
    ) = stream::iter(fcs)
        .then(|(p, fc)| {
            cloned!(sm_exp_data);
            async move {
                match &fc {
                    FileChange::Change(tfc) => match &tfc.file_type() {
                        FileType::GitSubmodule => expand_git_submodule_file_change(
                            ctx,
                            small_repo,
                            sm_exp_data.clone(),
                            parents,
                            p.clone(),
                            tfc.content_id(),
                        )
                        .timed()
                        .await
                        .log_future_stats(
                            ctx.scuba().clone(),
                            "Expand git submodule file change",
                            format!("submodule_file_path: {p}"),
                        ),
                        _ => {
                            if sm_exp_data.submodule_deps.contains_key(&p) {
                                // A normal file is replacing a submodule in the
                                // small repo, which means that the submodule
                                // expansion is being implicitly deleted in the
                                // large repo.
                                // If the expansion is deleted, we also need to
                                // delete the submodule metadata file.
                                let x_repo_sm_metadata_path =
                                    get_x_repo_submodule_metadata_file_path(
                                        &SubmodulePath(p.clone()),
                                        sm_exp_data.x_repo_submodule_metadata_file_prefix,
                                    )?;
                                return Ok(vec![
                                    ExpansionFileChange::Original((p.clone(), fc)),
                                    // Explicit deletion for the submodule
                                    // metadata file
                                    ExpansionFileChange::Generated((
                                        x_repo_sm_metadata_path,
                                        FileChange::Deletion,
                                        // Make the path optional for deletion
                                        SubmodulePath(p.clone()),
                                    )),
                                ]);
                            };
                            Ok(vec![ExpansionFileChange::Original((p, fc))])
                        }
                    },
                    FileChange::Deletion => {
                        handle_submodule_deletion(ctx, small_repo, sm_exp_data, parents, p).await
                    }
                    _ => Ok(vec![ExpansionFileChange::Original((p, fc))]),
                }
            }
        })
        .try_collect::<Vec<_>>()
        .await?
        .into_iter()
        .flatten()
        .try_fold(
            (
                SortedVectorMap::<NonRootMPath, FileChange>::new(),
                SortedVectorMap::<NonRootMPath, FileChange>::new(),
                HashMap::<SubmodulePath, HashSet<ContentId>>::new(),
            ),
            |(mut orig_fcs, mut exp_fcs, mut submodule_expansion_content_ids), exp_fc| {
                match exp_fc {
                    // If this file change came from the original bonsai, always
                    // insert it into the final file changes map.
                    ExpansionFileChange::Original((p, fc)) => {
                        orig_fcs.insert(p, fc);
                        Ok((orig_fcs, exp_fcs, submodule_expansion_content_ids))
                    }
                    // If this file change was generated by the submodule expansion
                    // primitives, only add it to the file changes map if there
                    // isn't a a file change for the same path, which would have
                    // come from the original bonsai.
                    ExpansionFileChange::Generated((p, fc, sm_path)) => {
                        // Make sure that no submodule file changes go through to the large repo
                        let (path, file_change) = match &fc {
                            FileChange::Change(tfc) => {
                                ensure!(
                                    tfc.file_type() != FileType::GitSubmodule,
                                    "Submodule file type should not be present in the file changes"
                                );
                                (p, fc.clone())
                            }
                            _ => (p, fc.clone()),
                        };
                        match exp_fcs.get(&path) {
                            Some(existing_fc) => {
                                // If there's already a generated file change for
                                // that path, check which one is a deletion
                                // and which one is a change/addition.
                                // The latter should have precedence over
                                // deletions.
                                // There should never be two file changes of the same
                                // type in the same path, crash in this unexpected
                                // scenario.
                                match (existing_fc.is_removed(), file_change.is_removed()) {
                                    // New file change should replace the deletion
                                    (true, false) => exp_fcs.insert(path, file_change),
                                    // New file change is a deletion, so keep the
                                    // existing one
                                    (false, true) => None,
                                    // UNEXPECTED: we should never generate two
                                    // deletions for the same path
                                    (true, true) => {
                                        return Err(anyhow!(
                                            "Two Deletions were generated for the same path"
                                        ));
                                    }
                                    // UNEXPECTED: we should never generate two
                                    // different changes for the same path
                                    (false, false) => {
                                        return Err(anyhow!(
                                            "Two file changes were generated for the same path"
                                        ));
                                    }
                                };
                            }
                            None => {
                                exp_fcs.insert(path, file_change);
                            }
                        };

                        if let Some(content_id) = fc.simplify().map(|fc| fc.content_id()) {
                            submodule_expansion_content_ids
                                .entry(sm_path)
                                .or_insert_with(HashSet::new)
                                .insert(content_id);
                        };
                        Ok((orig_fcs, exp_fcs, submodule_expansion_content_ids))
                    }
                }
            },
        )?;
    // File changes from the original bonsai should always take precedence over
    // changes generated by the expansion primitives.
    expanded_fcs.extend(orig_fcs);

    let new_cs = BonsaiChangesetMut {
        file_changes: expanded_fcs,
        ..cs
    };

    let new_cs = new_cs.freeze()?;

    Ok((new_cs.into_mut(), submodule_expansion_content_ids))
}

/// Expand a single file change from a git submodule.
///
/// In the source repo, the git submodule is a file containing the git hash of
/// the submodule's commit that the source repo depends on (let's call this
/// commit `X`).
///
/// In the target repo, the submodule path will be a directory containing the
/// contents of the submodule's working copy at commit `X`.
///
/// **EXAMPLE:** let's consider repos `source` and `A`, where `A` is a git submodule
/// dependency of `source` mirrored in Mononoke.
/// If A has commits X-Y-Z, a file change being expanded here would be, for example,
/// modifying the contents of the submodule file from `X` to `Z`.
///
/// This function will generate all the file changes to bring the working copy
/// of subdirectory `A` in `source` to the working copy of commit `Z` in `A`.
///
/// **IMPORTANT NOTES**
///
/// This function assumes that all the submodules (direct or recursive) are
/// mirrored in a Mononoke repo and these repos are loaded and available inside
/// `submodule_deps`. It will crash if that's not the case.
///
/// This depends on fsnodes from the commits in the source repo and the
/// submodule repos, so if they aren't already derived, they will be during the
/// expansion process.
///
/// All the file content blobs from all the submodule repos will be copied into
/// the source repo's blobstore, so that the commit rewrite crate can copy them
/// into the target repo.
async fn expand_git_submodule_file_change<'a, R: Repo>(
    ctx: &'a CoreContext,
    small_repo: &'a R,
    sm_exp_data: SubmoduleExpansionData<'a, R>,
    parents: &'a [ChangesetId],
    submodule_file_path: NonRootMPath,
    submodule_file_content_id: ContentId,
) -> Result<Vec<ExpansionFileChange>> {
    let submodule_path = SubmodulePath(submodule_file_path);
    // Contains lists of file changes along
    // with the submodule these file changes are
    // from, so that the file content blobs are
    // copied from each submodule's blobstore into
    // the source repo's blobstore.
    let exp_results = expand_git_submodule(
        ctx,
        small_repo,
        parents,
        submodule_path.clone(),
        sm_exp_data.clone(),
        submodule_file_content_id,
    )
    .await?;

    // Build the list of file changes to be returned
    let expanded_file_changes = exp_results
        .into_iter()
        .flat_map(move |(sm_path, fcs)| {
            fcs.into_iter()
                .map(move |(p, fc)| ExpansionFileChange::Generated((p, fc, sm_path.clone())))
        })
        .collect();

    anyhow::Ok(expanded_file_changes)
}

#[async_recursion]
async fn expand_git_submodule<'a, R>(
    ctx: &'a CoreContext,
    small_repo: &'a R,
    // Parents from the **source repo commit** being rewritten.
    // This is needed to get the hash of the previous commit of the submodule
    // being expanded.
    parents: &'a [ChangesetId],
    // Path of the submodule file in the source repo, which contains the encoded
    // git hash of the submodule's commit that the source repo depends on.
    submodule_path: SubmodulePath,
    sm_exp_data: SubmoduleExpansionData<'a, R>,
    // The
    submodule_file_content_id: ContentId,
    // Returns a map from submodule path to a list of file changes, so that
    // before the file changes are rewritten, the file content blobs are copied
    // from the appropriate submodule repo into the source repo's blobstore.
) -> Result<HashMap<SubmodulePath, Vec<(NonRootMPath, FileChange)>>>
where
    R: Repo,
{
    log_debug(ctx, format!("Expanding submodule {}", &submodule_path));

    let submodule_repo = get_submodule_repo(&submodule_path, sm_exp_data.submodule_deps)?;
    let git_submodule_sha1 =
        get_git_hash_from_submodule_file(ctx, small_repo, submodule_file_content_id).await?;

    log_debug(
        ctx,
        format!(
            "submodule_path: {} | git_submodule_hash: {} | submodule_repo name: {}",
            &submodule_path,
            &git_submodule_sha1,
            &submodule_repo.repo_identity().name()
        ),
    );

    let sm_changeset_id = get_submodule_bonsai_changeset_id(
        ctx,
        submodule_repo,
        git_submodule_sha1,
        &sm_exp_data.dangling_submodule_pointers,
    )
    .await?;

    let sm_parents = get_previous_submodule_commits(
        ctx,
        parents,
        small_repo,
        submodule_path.clone(),
        submodule_repo,
        &sm_exp_data.dangling_submodule_pointers,
    )
    .await?;

    // `sm_file_changes` are the changes in the submodule being processed
    // that should be expanded.
    //  `recursive_sm_file_changes` are the changes from all submodules that
    // the current submodule depends on.
    // The latter need to be stored separately because all the file content
    // blobs will need to be copied from the appropriate repository after
    // generating the file changes.
    let (sm_file_changes, recursive_sm_file_changes) =
        submodule_diff(ctx, submodule_repo, sm_changeset_id, sm_parents)
            .await?
            .map_ok(|diff| {
                cloned!(submodule_path, sm_exp_data);

                async move {
                    match diff {
                        BonsaiDiffFileChange::Changed(path, (file_type, content_id, size))
                        | BonsaiDiffFileChange::ChangedReusedId(
                            path,
                            (file_type, content_id, size),
                        ) => {
                            if file_type != FileType::GitSubmodule {
                                // Non-submodule file changes just need to have the submodule
                                // path in the source repo prepended to their path.
                                let new_tfc = TrackedFileChange::new(
                                    content_id,
                                    file_type,
                                    size,
                                    None,
                                    GitLfs::FullContent,
                                );
                                let path_in_sm = submodule_path.0.join(&path);
                                let mut fcs =
                                    vec![(path_in_sm.clone(), FileChange::Change(new_tfc))];

                                // This handles the scenario where a submodule is replaced by a
                                // file which results in implicit submodule deletion. The only
                                // extra thing we need to do to handle this scenario properly is to
                                // remove the corresponding submodule metadata file.
                                if sm_exp_data.submodule_deps.contains_key(&path_in_sm) {
                                    let x_repo_sm_metadata_path =
                                        get_x_repo_submodule_metadata_file_path(
                                            &SubmodulePath(path_in_sm),
                                            sm_exp_data.x_repo_submodule_metadata_file_prefix,
                                        )?;
                                    fcs.push((x_repo_sm_metadata_path, FileChange::Deletion));
                                }

                                return Ok(Left(fcs));
                            }

                            let previous_submodule_commits = get_previous_submodule_commits(
                                ctx,
                                parents,
                                small_repo,
                                submodule_path.clone(),
                                submodule_repo,
                                &sm_exp_data.dangling_submodule_pointers,
                            )
                            .await?;

                            process_recursive_submodule_file_change(
                                ctx,
                                // Use the previous commits of the submodule as parents
                                // when expanding any recursive submodules.
                                previous_submodule_commits.as_slice(),
                                submodule_path,
                                submodule_repo,
                                sm_exp_data,
                                SubmodulePath(path),
                                content_id,
                            )
                            .await
                        }
                        BonsaiDiffFileChange::Deleted(path) => {
                            let previous_submodule_commits = get_previous_submodule_commits(
                                ctx,
                                parents,
                                small_repo,
                                submodule_path.clone(),
                                submodule_repo,
                                &sm_exp_data.dangling_submodule_pointers,
                            )
                            .await?;

                            let rec_small_repo_deps = build_recursive_submodule_deps(
                                sm_exp_data.submodule_deps,
                                &submodule_path.0,
                            );

                            let rec_sm_exp_data = SubmoduleExpansionData {
                                submodule_deps: &rec_small_repo_deps,
                                ..sm_exp_data
                            };

                            let paths_to_delete = handle_submodule_deletion(
                                ctx,
                                submodule_repo,
                                rec_sm_exp_data,
                                previous_submodule_commits.as_slice(),
                                path,
                            )
                            .await?
                            .into_iter()
                            .map(ExpansionFileChange::into_path)
                            .collect::<Vec<_>>();

                            let fcs = paths_to_delete
                                .into_iter()
                                .map(|p| {
                                    cloned!(submodule_path);
                                    let full_path = submodule_path.0.join(&p);
                                    (full_path, FileChange::Deletion)
                                })
                                .collect::<Vec<_>>();

                            Ok(Left(fcs))
                        }
                    }
                }
            })
            .map_err(|e| anyhow!("Failed to generate a BonsaiDiffFileChange: {e}"))
            .try_buffer_unordered(100)
            .try_collect::<Vec<_>>()
            .await?
            .into_iter()
            .fold((vec![], HashMap::new()), |(mut fcs, mut sm_fcs), res| {
                match res {
                    Left(res_fcs) => {
                        fcs.extend(res_fcs);
                    }
                    Right(sm_fcs_map) => {
                        sm_fcs_map.into_iter().for_each(|(path, res_fcs)| {
                            let fcs = sm_fcs.entry(path).or_insert(vec![]);
                            fcs.extend(res_fcs);
                        });
                    }
                }

                (fcs, sm_fcs)
            });

    // File changes generated for the expanded submodule and changes to its
    // x-repo submodule metadata file
    let all_file_changes = generate_additional_file_changes(
        ctx,
        small_repo,
        parents,
        submodule_path.clone(),
        submodule_file_content_id,
        sm_exp_data,
        sm_file_changes,
    )
    .await?;

    let mut fc_map = hashmap![submodule_path => all_file_changes];
    recursive_sm_file_changes
        .into_iter()
        .for_each(|(sm_path, sm_fcs)| {
            let fcs = fc_map.entry(sm_path).or_insert(vec![]);
            fcs.extend(sm_fcs);
        });

    Ok(fc_map)
}

/// Small helper function mostly to avoid confusion with nomenclature.
///
/// Consider the following repos `source -> A -> B`, where `source` is the repo
/// being exported, `A` is a direct submodule dependency and `B` is a submodule
/// inside repo `A`.
/// This will generate the file changes to expand `B` inside `A` and will
/// then modify them so they're placed inside the copy of `A` in `source`.
async fn process_recursive_submodule_file_change<'a, R: Repo>(
    ctx: &'a CoreContext,
    // Parents that should be used to generate the proper delta file changes.
    parents: &'a [ChangesetId],
    // Path of submodule `A` within repo `source`.
    submodule_path: SubmodulePath,
    submodule_repo: &'a R,
    sm_exp_data: SubmoduleExpansionData<'a, R>,
    // Path of submodule `B` within repo `A`.
    recursive_submodule_path: SubmodulePath,
    // Content id of the submodule `B` file inside repo `A`.
    // It contains the git hash of the `B` commit that `A` depends on.
    recursive_sm_file_content_id: ContentId,
) -> Result<
    Either<
        Vec<(NonRootMPath, FileChange)>,
        HashMap<SubmodulePath, Vec<(NonRootMPath, FileChange)>>,
    >,
> {
    // Create a new source_deps_map for the recursive call, removing the
    // submodule prefix from the keys that have it and ignoring the ones that
    // don't, because they're not relevant to the submodule being processed.
    // This prefix is added back to the results, before they're returned.
    let rec_small_repo_deps =
        build_recursive_submodule_deps(sm_exp_data.submodule_deps, &submodule_path.0);

    let sm_exp_data = SubmoduleExpansionData {
        submodule_deps: &rec_small_repo_deps,
        ..sm_exp_data
    };

    let rec_sm_file_changes = expand_git_submodule(
        ctx,
        submodule_repo,
        parents,
        recursive_submodule_path,
        sm_exp_data,
        recursive_sm_file_content_id,
    )
    .await
    .with_context(|| {
        format!(
            "Failed to expand recursive submodule inside submodule {}",
            &submodule_repo.repo_identity().name()
        )
    })?;

    let sm_fcs_map = rec_sm_file_changes
        .into_iter()
        .map(|(rec_sm, rec_fcs)| {
            let mod_fcs = rec_fcs
                .into_iter()
                .map(|(p, fc)| (submodule_path.0.join(&p), fc))
                .collect::<Vec<_>>();
            // Add back the prefix of the root submodule to the recursive one
            (SubmodulePath(submodule_path.0.join(&rec_sm.0)), mod_fcs)
        })
        .collect::<HashMap<_, _>>();
    Ok(Right(sm_fcs_map))
}

/// Get the previous commit of the submodule that the source repo depended on.
/// These changesets will be used to generate the best delta.
///
/// **Example:**
/// The source repo depends on submodule A. A has commits X-Y-Z.
/// The source repo initially depended on X, but now we're syncing a commit that
/// updates the submodule straight to Z.
///
/// When generating the delta (i.e. file changes to the submodule directory),
/// we should generate the differences between commit X and Z, instead of
/// copying the entire working copy in commit Z (which is a lot of unnecessary work).
///
/// Using the example from above, the output of this function would be `[X]`.
async fn get_previous_submodule_commits<'a, R: Repo>(
    ctx: &'a CoreContext,
    // Parents of the changeset being synced from the source repo. We get the
    // contents of the submodule file in those revisions to get the previous
    // submodule commit the source repo depended on.
    parents: &'a [ChangesetId],
    small_repo: &'a R,
    // Path of submodule `A` within repo `source`.
    submodule_path: SubmodulePath,
    // Submodule repo in Mononoke
    submodule_repo: &'a R,
    dangling_submodule_pointers: &[GitSha1],
) -> Result<Vec<ChangesetId>> {
    let parents_vec = parents
        .iter()
        .map(|cs_id| anyhow::Ok(*cs_id))
        .collect::<Vec<_>>();

    // Get the changeset ids of the previous revision of the submodule that the
    // source repo depended on, if the submodule is being updated. If the
    // submodule is being added, this set will be empty.
    let sm_parents: Vec<ChangesetId> = stream::iter(parents_vec)
        .try_filter_map(|cs_id| {
            cloned!(ctx, submodule_path, dangling_submodule_pointers);

            async move {
                // Check the submodule path on that revision
                match get_submodule_file_content_id(&ctx, small_repo, cs_id, &submodule_path.0)
                    .await?
                {
                    // If it's a submodule file, the submodule is being updated
                    Some(submodule_file_content_id) => {
                        // File is a submodule, so get the git hash that it stored
                        // which represents the pointer to that submodule.
                        let git_sha1 = get_git_hash_from_submodule_file(
                            &ctx,
                            small_repo,
                            submodule_file_content_id,
                        )
                        .await?;

                        // From the git hash, get the bonsai changeset it in the
                        // submodule Mononoke repo.
                        let sm_parent_cs_id = get_submodule_bonsai_changeset_id(
                            &ctx,
                            submodule_repo,
                            git_sha1,
                            dangling_submodule_pointers,
                        )
                        .await?;
                        Ok(Some(sm_parent_cs_id))
                    }
                    // If it doesn't exist, or is a directory, skip it because
                    // it's not a revision that can be used as a parent to
                    // generate delta for the submodule expansion.  If it is a
                    // file of type other than GitSubmodule, it means that a
                    // submodule is being added in the place of a regular
                    // file, so this revision didn't have a dependency on the
                    // submodule, and can also be skipped.
                    None => Ok(None),
                }
                // Get content id of the file
            }
        })
        .try_collect::<Vec<_>>()
        .await?;
    Ok(sm_parents)
}

/// If the path being deleted is a file of type GitSubmodule, then we should
/// delete its entire expansion and its metadata file in the large repo.
async fn handle_submodule_deletion<'a, R: Repo>(
    ctx: &'a CoreContext,
    small_repo: &'a R,
    sm_exp_data: SubmoduleExpansionData<'a, R>,
    parents: &'a [ChangesetId],
    deleted_path: NonRootMPath,
) -> Result<Vec<ExpansionFileChange>> {
    // If the path is in the submodule_deps keys, it's almost
    // certainly a submodule being deleted.
    if sm_exp_data.submodule_deps.contains_key(&deleted_path) {
        // However, to be certain, let's verify that this file
        // was indeed of type `GitSubmodule` by getting the checking
        // the FileType of the submodule file path in each of the parents.
        let is_git_submodule_file = stream::iter(parents)
            .map(|cs_id| is_path_git_submodule(ctx, small_repo, *cs_id, &deleted_path))
            .buffer_unordered(10)
            .boxed()
            .try_collect::<Vec<_>>()
            .await?
            .into_iter()
            .any(|is_git_submodule| is_git_submodule);

        // Not a submodule file, just a file at the same path
        // where a git submodule used to be, so just delete the file normally.
        if !is_git_submodule_file {
            return Ok(vec![ExpansionFileChange::Original((
                deleted_path,
                FileChange::Deletion,
            ))]);
        }

        // This is a submodule file, so delete its entire expanded directory.
        let generated_deletions =
            delete_submodule_expansion(ctx, small_repo, sm_exp_data, parents, deleted_path.clone())
                .timed()
                .await
                .log_future_stats(
                    ctx.scuba().clone(),
                    "Deleting submodule expansion",
                    format!("Submodule path: {deleted_path}"),
                )?;

        return Ok(generated_deletions
            .into_iter()
            .map(|p| {
                ExpansionFileChange::Generated((
                    p,
                    FileChange::Deletion,
                    SubmodulePath(deleted_path.clone()),
                ))
            })
            .collect());
    };

    Ok(vec![ExpansionFileChange::Original((
        deleted_path,
        FileChange::Deletion,
    ))])
}

/// After confirming that the path being deleted is indeed a submodule file,
/// generate the deletion for its entire expanded directory.
///
/// The files to be deleted are generated by expanding the submodule from the
/// parents' changesets as if they had no parents, i.e. expanding the pointer
/// from the parents are if they were being added for the first time.
async fn delete_submodule_expansion<'a, R: Repo>(
    ctx: &'a CoreContext,
    small_repo: &'a R,
    sm_exp_data: SubmoduleExpansionData<'a, R>,
    parents: &'a [ChangesetId],
    submodule_file_path: NonRootMPath,
) -> Result<Vec<NonRootMPath>> {
    let submodule_path = SubmodulePath(submodule_file_path.clone());

    if parents.len() > 1 {
        log_info(
            ctx,
            format!(
                "Deleting submodule expansion for {} with multiple parents {parents:#?}",
                submodule_path.0
            ),
        );
    }

    let sm_file_content_ids = stream::iter(parents).map(anyhow::Ok).try_filter_map(|cs_id| {
        cloned!(submodule_path);
        async move {
            get_submodule_file_content_id(ctx, small_repo, *cs_id, &submodule_path.0)
                .await
                .with_context(|| {
                    log_info(ctx, format!("Parent changeset {cs_id}, submodule_path: {0}, small_repo: {1}", submodule_path.0, small_repo.repo_identity().name()));    
                    anyhow!("Failed to get submodule file content id from parent changeset for submodule deletion")
                })
        }
    }).try_collect::<Vec<_>>().await?;

    if sm_file_content_ids.is_empty() {
        // At least one of the parents should have the submodule in its working
        // copy. This file change will provide the submodule commit that is
        // currently expanded.
        log_info(
            ctx,
            format!(
                "Parents: {parents:#?}, submodule_path: {0}, small_repo: {1}",
                submodule_path.0,
                small_repo.repo_identity().name()
            ),
        );
        return Err(anyhow!(
            "Didn't find content id of submodule file in any of the parents."
        ));
    }

    // Get the entire working copy of the submodule in those revisions, so we
    // can generate the proper paths to be deleted.
    let full_expansion_paths = stream::iter(sm_file_content_ids)
        .map(anyhow::Ok)
        .try_fold(
            Vec::new(),
            |mut full_exp_paths, submodule_file_content_id| {
                cloned!(sm_exp_data, submodule_file_path);
                async move {
                    let expansion_changes = expand_git_submodule_file_change(
                        ctx,
                        small_repo,
                        sm_exp_data,
                        &[], // Get the entire expansion by passing an empty set of parents
                        submodule_file_path,
                        submodule_file_content_id,
                    )
                    .await?;
                    let paths = expansion_changes
                        .into_iter()
                        .map(ExpansionFileChange::into_path)
                        .collect::<Vec<_>>();
                    full_exp_paths.extend(paths);
                    anyhow::Ok(full_exp_paths)
                }
            },
        )
        .await?;

    Ok(full_expansion_paths)
}

/**
 After getting the file changes from the submodule repo, generate any additional
 file changes needed to bring the bonsai into a healthy/consistent state.
 - Submodule metadata file, which stores the pointer to the submodule revision
   being expanded and is used to validate consistency between the revision and
   its expansion.
 - Deletions of files/directories that are being replaced by the creation of the
   submodule expansion.
*/
async fn generate_additional_file_changes<'a, R: Repo>(
    ctx: &'a CoreContext,
    small_repo: &'a R,
    parents: &'a [ChangesetId],
    submodule_path: SubmodulePath,
    submodule_file_content_id: ContentId,
    sm_exp_data: SubmoduleExpansionData<'a, R>,
    expanded_file_changes: Vec<(NonRootMPath, FileChange)>,
) -> Result<Vec<(NonRootMPath, FileChange)>> {
    // Step 1: Generate the submodule metadata file change

    // After expanding the submodule, we also need to generate the x-repo
    // submodule metadata file, to keep track of the git hash that this expansion
    // corresponds to.
    let x_repo_sm_metadata_path = get_x_repo_submodule_metadata_file_path(
        &submodule_path,
        sm_exp_data.x_repo_submodule_metadata_file_prefix,
    )?;

    let git_submodule_sha1 =
        get_git_hash_from_submodule_file(ctx, small_repo, submodule_file_content_id).await?;

    // Create the x-repo submodule metadata file change, containing the decoded
    // hash of the git commit from the submodule that's being expanded.
    let metadata_file_content = FileContents::new_bytes(git_submodule_sha1.to_string());
    let metadata_file_size = metadata_file_content.size();

    let submodule_repo = get_submodule_repo(&submodule_path, sm_exp_data.submodule_deps)?;

    // Store this metadata file content in the submodule repo's blobstore, so
    // it can be copied to the small repo's blobstore afterwards.
    let metadata_file_content_id = metadata_file_content
        .clone()
        .into_blob()
        .store(ctx, submodule_repo.repo_blobstore())
        .await?;

    // The submodule metadata file content has to be in the large repo's
    // blobstore for validation. All these writes only happen in memory and are
    // not persisted to the large repo's blobstore until the bonsai is actually
    // synced and committed there.
    let _ = metadata_file_content
        .clone()
        .into_blob()
        .store(ctx, sm_exp_data.large_repo.repo_blobstore())
        .await?;

    // The metadata file will have the same content as the submodule file
    // change in the source repo, but it will be a regular file, because in
    // the large repo we can never have file changes of type `GitSubmodule`.
    let x_repo_sm_metadata_fc = FileChange::tracked(
        metadata_file_content_id,
        FileType::Regular,
        metadata_file_size,
        None,
        GitLfs::FullContent,
    );
    let mut all_changes = vec![(x_repo_sm_metadata_path, x_repo_sm_metadata_fc)];

    // Step 2: Generate the deletions of files/directories that are being
    // replaced by the creation of the submodule expansion.

    // Get the non submodule files underneath the submodule path in the parent commits
    let non_submodule_paths: Vec<NonRootMPath> = stream::iter(parents)
        .then(|cs_id| {
            // If there is an entry for a **non-GitSubmodule file type**, it
            // means that in the large repo we're replacing a regular file or
            // regular directory with the submodule expansion, so we need to
            // generate deletions for it.
            // The paths to be deleted, will be all the leaves under the
            // submodule path.
            list_non_submodule_files_under(ctx, small_repo, *cs_id, submodule_path.clone())
        })
        .boxed()
        .try_flatten()
        .try_collect::<Vec<_>>()
        .await?;

    // NOTE: Deletions must be added first, because the expanded changes take
    // precedence over the deletions. i.e. if we generate a Deletion and a Change
    // for the same path, it means that the submodule expansion is replacing
    // a regular directory with a file in the same path.
    // We don't want this file actually deleted.

    all_changes.extend(
        non_submodule_paths
            .into_iter()
            .map(|path| (path, FileChange::Deletion)),
    );

    all_changes.extend(expanded_file_changes);

    Ok(all_changes)
}

impl ExpansionFileChange {
    pub fn into_inner(self) -> (NonRootMPath, FileChange) {
        match self {
            Self::Original(path_fc) => path_fc,
            Self::Generated((path, fc, _sm_cids)) => (path, fc),
        }
    }

    pub fn into_path(self) -> NonRootMPath {
        self.into_inner().0
    }
}
